package org.example.batch.config;

import org.example.batch.entity.Person;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.transaction.PlatformTransactionManager;

import javax.annotation.Resource;
import javax.sql.DataSource;

@Configuration
@EnableBatchProcessing
public class BatchConfig extends DefaultBatchConfigurer {

    //private final DataSource batchDataSource;

    @Resource
    private JobBuilderFactory jobBuilderFactory;

    @Resource
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private DataSource businessDataSource; // 业务数据源

    @Autowired
    private DataSource batchDataSource; // 批处理数据源

    /*public BatchConfig(@Qualifier("batchDataSource") DataSource batchDataSource) {
        this.batchDataSource = batchDataSource;
    }*/

    @Override
    public void setDataSource(DataSource dataSource) {
        // 不调用超类方法以避免覆盖我们为Batch指定的数据源
    }

    @Override
    protected JobRepository createJobRepository() throws Exception {
        System.out.println(batchDataSource.getConnection().getMetaData().getURL());
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(batchDataSource);
        factory.setTransactionManager(batchTransactionManager());
        factory.afterPropertiesSet();
        return factory.getObject();
    }

    @Bean
    public PlatformTransactionManager batchTransactionManager() {
        return new ResourcelessTransactionManager();
    }


    @Bean
    public JdbcCursorItemReader<Person> reader() {
        JdbcCursorItemReader<Person> reader = new JdbcCursorItemReader<>();
        reader.setDataSource(businessDataSource);
        reader.setSql("SELECT name FROM org_person");
       // reader.setRowMapper(new YourEntityRowMapper());
        reader.setRowMapper(new BeanPropertyRowMapper<>(Person.class));
        return reader;
    }

    @Bean
    public ItemProcessor<Person, Person> processor() {
        return item -> {
            // 对数据进行处理
            return item;
        };
    }

    @Bean
    public ItemWriter<Person> writer() {
       /* JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        writer.setSql("INSERT INTO target_table (col1, col2) VALUES (:col1, :col2)");
        writer.setDataSource(batchDataSource);
        return writer;*/
        return new CustomJdbcBatchItemWriter(businessDataSource);
    }

    @Bean
    public Step copyDataStep() {
        return stepBuilderFactory.get("copyDataStep")
                .<Person, Person>chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

    @Bean
    public Job processJob() {
        return jobBuilderFactory.get("processJob")
                .incrementer(new RunIdIncrementer()) // 每次运行生成新的run id
                .start(copyDataStep()) // 添加你的step
                .build();
    }
}